package com.atguigu.gmall.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.app.function.DimAsyncFunction;
import com.atguigu.gmall.bean.TradeProvinceOrderWindow;
import com.atguigu.gmall.util.DateFormatUtils;
import com.atguigu.gmall.util.MyClickHouseUtils;
import com.atguigu.gmall.util.MyKafkaUtils;
import com.atguigu.gmall.util.TimestampLtz3CompareUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * 交易域省份粒度下单各窗口汇总表
 * 数据流：Web/app -> nginx -> 业务服务器(Mysql) -> Maxwell -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp -> Kafka(DWD)
 * -> FlinkApp -> ClickHouse(DWS)
 *
 * 程  序：Mock  ->  Mysql  ->  Maxwell -> Kafka(ZK)  ->  DwdTradeOrderPreProcess -> Kafka(ZK) -> DwdTradeOrderDetail -> Kafka(ZK)
 * -> DwsTradeProvinceOrderWindow(Phoenix-(HBase-HDFS、ZK)、Redis) -> ClickHouse(ZK)
 *
 * @author : ranzlupup
 * @since : 2023/6/6 14:12
 */
public class DwsTradeProvinceOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); //生产环境中设置为Kafka主题的分区数

        //1.1 开启CheckPoint
        //env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        //env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        //env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));

        //1.2 设置状态后端
        //env.setStateBackend(new HashMapStateBackend());
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/211126/ck");
        //System.setProperty("HADOOP_USER_NAME", "atguigu");

        //1.3 设置状态的TTL  生产环境设置为最大乱序程度
        //tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        //TODO 2.从Kafka订单明细主题读取数据
        String topicName = "FLINK_DWD_TRADE_ORDER_DETAIL";
        String groupId = "dws_trade_province_order_window_211126";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtils.getFlinkKafkaConsumer(topicName, groupId);
        DataStreamSource<String> kafkaDS = env.addSource(kafkaConsumer);

        //TODO 3.转换数据结构
        SingleOutputStreamOperator<JSONObject> jsonObjectDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    System.out.println("脏数据: " + value);
                }

            }
        });

        //TODO 4.按照唯一键去重
        KeyedStream<JSONObject, String> keyedStream = jsonObjectDS.keyBy(
                new KeySelector<JSONObject, String>() {
                    @Override
                    public String getKey(JSONObject value) throws Exception {
                        return value.getString("id");
                    }
                });

        //! 设置定时器，定时器触发时，窗口关闭，收集窗口数据
        //! 要用到定时器，所以得用process
        SingleOutputStreamOperator<JSONObject> processFilterDS = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            private ValueState<JSONObject> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                valueState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("value-state", JSONObject.class));
            }

            @Override
            public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                // 取出状态里面的数据
                JSONObject lastValue = valueState.value();

                // 判断状态数据是否为null
                if (lastValue == null) {
                    // 将数据存入状态，并且设置定时器
                    valueState.update(value);
                    long processingTime = ctx.timerService().currentProcessingTime();
                    ctx.timerService().registerProcessingTimeTimer(processingTime + 5000L);
                } else {
                    // 如果不为null，取出状态里面的时间与value中的时间比较，保留时间晚的数据
                    String lastTs = lastValue.getString("row_op_ts");
                    String curTs = value.getString("row_op_ts");

                    if (TimestampLtz3CompareUtil.compare(lastTs, curTs) != 1) {
                        valueState.update(value);
                    }
                }
            }

            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                //! 定时器触发，输出数据并清空状态
                JSONObject lastValue = valueState.value();
                out.collect(lastValue);
                valueState.clear();
            }
        });

        //TODO 5.转换数据结构（转换成Bean类型）
        SingleOutputStreamOperator<TradeProvinceOrderWindow> tradeProvinceOrderWindowDS = processFilterDS.map(
                new MapFunction<JSONObject, TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow map(JSONObject value) throws Exception {
                        // 用Set存储所有的orderId
                        HashSet<String> orderIdSet = new HashSet<>();
                        orderIdSet.add(value.getString("order_id"));
                        return TradeProvinceOrderWindow
                                .builder()
                                .provinceId(value.getString("province_id"))
                                .orderCount(0L)
                                .orderIdSet(orderIdSet)
                                .orderAmount(value.getDouble("split_total_amount"))
                                .ts(DateFormatUtils.toTs(value.getString("create_time"), true))
                                .build();
                    }
                }
        );

        //TODO 6.提取时间戳生成WaterMark
        SingleOutputStreamOperator<TradeProvinceOrderWindow> tradeProvinceOrderWindowWithWmDS = tradeProvinceOrderWindowDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderWindow>() {
                            @Override
                            public long extractTimestamp(TradeProvinceOrderWindow element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
        );

        //TODO 7.按照省份ID分组 province_id
        KeyedStream<TradeProvinceOrderWindow, String> tradeProvinceOrderWindowStringKeyedStream = tradeProvinceOrderWindowWithWmDS.keyBy(TradeProvinceOrderWindow::getProvinceId);

        //TODO 8.开窗
        WindowedStream<TradeProvinceOrderWindow, String, TimeWindow> windowedStream = tradeProvinceOrderWindowStringKeyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        //TODO 9.聚合
        SingleOutputStreamOperator<TradeProvinceOrderWindow> reduceDS = windowedStream.reduce(

                new ReduceFunction<TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow value1, TradeProvinceOrderWindow value2) throws Exception {
                        Set<String> orderIdSet = value1.getOrderIdSet();
                        orderIdSet.addAll(value2.getOrderIdSet());
                        value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                        return value1;
                    }
                },
                new WindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeProvinceOrderWindow> input, Collector<TradeProvinceOrderWindow> out) throws Exception {
                        TradeProvinceOrderWindow tradeProvinceOrderWindow = input.iterator().next();

                        tradeProvinceOrderWindow.setTs(System.currentTimeMillis());

                        // 计算订单数量
                        tradeProvinceOrderWindow.setOrderCount((long) tradeProvinceOrderWindow.getOrderIdSet().size());
                        tradeProvinceOrderWindow.setStt(DateFormatUtils.toYmdHms(window.getStart()));
                        tradeProvinceOrderWindow.setEdt(DateFormatUtils.toYmdHms(window.getEnd()));

                        out.collect(tradeProvinceOrderWindow);
                    }
                }
        );
        reduceDS.print("reduceDS>>>>>>> ");

        //TODO 10.关联省份维度表补充省份名称字段
        SingleOutputStreamOperator<TradeProvinceOrderWindow> resultDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeProvinceOrderWindow>("DIM_BASE_PROVINCE") {
                    @Override
                    public String getKey(TradeProvinceOrderWindow input) {
                        return input.getProvinceId();
                    }

                    @Override
                    public void join(TradeProvinceOrderWindow input, JSONObject dimInfo) {
                        input.setProvinceName(dimInfo.getString("NAME"));

                    }
                },
                100,
                TimeUnit.SECONDS);

        resultDS.print("resultDS>>>>>>>>>>>> ");


        //TODO 11.将数据写出到ClickHouse
        resultDS.addSink(MyClickHouseUtils.getSinkFunction("insert into dws_trade_province_order_window values(?,?,?,?,?,?,?)"));

        //TODO 12.执行任务
        env.execute();
    }
}
